package com.shujia.stream

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

import scala.collection.mutable.ListBuffer

object Demo6ConList {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")


    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))


    /**
      * 读取电信行为数据
      *
      */

    val dianxin: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)


    /**
      * transform， foreachRDD  的外面属于Driver端，只执行一次
      * transform， foreachRDD  的内部算子的外部属于Driver端，买一个batch都会执行一次
      * 算子内属于Executor端，每一条数据执行一次
      */

    val filterDS: DStream[String] = dianxin.transform(rdd=>{


      //查询mysql 获取布控列表

      Class.forName("com.mysql.jdbc.Driver")
      //创建链接
      val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")

      //查询数据

      val stat: PreparedStatement = con.prepareStatement("select mdn from list")
      val resultSet: ResultSet = stat.executeQuery()

      val list = new ListBuffer[String]
      while (resultSet.next()) {
        val m: String = resultSet.getString("mdn")
        list += m
      }

      con.close()

      println(s"布控列表：$list")

      //将布控列表广播出去
      val broList: Broadcast[ListBuffer[String]] = sc.broadcast(list)


      val filterRDD: RDD[String] = rdd.filter(line=>{
        val mdn: String = line.split(",")(0)

        //判断手机号是否出现
        broList.value.contains(mdn)
      })

      //清除广播变量
      broList.unpersist()

      filterRDD
    })


    filterDS.print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
